package com.bujunjie.study.realtime.dim.function;

import com.bujunjie.study.realtime.common.constant.FlinkConstant;
import com.bujunjie.study.realtime.common.enums.MaxwellOpEnum;
import com.bujunjie.study.realtime.common.util.HBaseUtil;
import com.bujunjie.study.realtime.common.util.RedisUtil;
import com.bujunjie.study.realtime.dim.model.vo.DimTableVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.util.Arrays;
import java.util.Map;

/**
 * <p> HBase 写入的方法</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/15 9:28
 */
@Slf4j
public class HBaseSinkFunction extends RichSinkFunction<Tuple2<Map<String, Object>, DimTableVO>> {

    private Connection hbaseConn;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hbaseConn);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public void invoke(Tuple2<Map<String, Object>, DimTableVO> tup, Context context) throws Exception {
        log.info("写入到 HBase 的二元组数据");
        Map<String, Object> dataMap = tup.f0;
        DimTableVO dimTable = tup.f1;
        String type = (String) dataMap.get("type");
        dataMap.remove("type");
        //获取操作的HBase表的表名
        String sinkTable = dimTable.getSinkTable();
        //获取 rowKey
        String rowKey = (String) dataMap.get(dimTable.getSinkRowKey());
        //判断对业务数据库维度表进行了什么操作
        if (MaxwellOpEnum.DELETE.getOp().equals(type)) {
            //从业务数据库维度表中做了删除操作  需要将HBase维度表中对应的记录也删除掉
            HBaseUtil.delRow(hbaseConn, FlinkConstant.HBASE_NAMESPACE, sinkTable, rowKey);
        } else {
            //如果不是delete，可能的类型有insert、update、bootstrap-insert，上述操作对应的都是向HBase表中put数据
            String sinkFamily = dimTable.getSinkFamily();
            HBaseUtil.putRow(hbaseConn, FlinkConstant.HBASE_NAMESPACE, sinkTable, rowKey, sinkFamily, dataMap);
        }
        // 如果维度表数据发生了变化，将Redis中缓存的数据清除掉
        if (Arrays.asList(MaxwellOpEnum.DELETE.getOp(), MaxwellOpEnum.UPDATE.getOp()).contains(type)) {
            String key = RedisUtil.getKey(sinkTable, rowKey);
            jedis.del(key);
        }
    }
}
